package com.bw.app.dws;

import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dws_buy_count {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String topic="dwd_order_pre";
        String groupId="buy_count_2114";
        tableEnv.executeSql("create table buy_count_table(\n" +
                "   `sku_id`  STRING , \n" +
                "   `user_id` STRING, \n" +
                "   `buy_counts`STRING, \n" +
                "   `order_id` STRING,\n" +
                "   `sku_name` STRING,\n" +
                "   `order_price` STRING,\n" +
                "   `order_status` STRING,\n" +
                "   `row_op_ts` STRING,\n" +
                "   `source_type_name` STRING,\n" +
                "  `time_ltz` AS TO_TIMESTAMP_LTZ(\n" +
                "    UNIX_TIMESTAMP(REPLACE(row_op_ts, ' ', 'T'), 'yyyy-MM-dd''T''HH:mm:ss') * 1000,\n" +
                "    3\n" +  // 毫秒精度
                "  ),\n" +
                " WATERMARK FOR time_ltz AS time_ltz  - INTERVAL '2' seconds" +
                ")"+ MyKafkaUtil.getKafkaDDL(topic,groupId));


        tableEnv.executeSql("with  t1  as( " +
                "select   *   from buy_count_table where order_status in('1002','1004','1005') " +
                ") SELECT " +
                "  window_start, " +
                "  window_end, sku_id,order_id,sku_name,order_price,source_type_name," +
                "  COUNT(DISTINCT user_id) AS buy_counts " +
                "FROM TABLE(" +  // 确保FROM后使用TABLE(包裹
                "  TUMBLE(TABLE buy_count_table, DESCRIPTOR(`time_ltz`), INTERVAL '10' SECOND)" +  // 可能需要反引号包裹字段名
                ") " +// 若过滤条件针对原始数据，使用WHERE
                "GROUP BY window_start, window_end,sku_id,order_id,sku_name,order_price,source_type_name").print();

        env.execute();

    }
}
